package com.gvsoft;

import com.gvsoft.analyse.*;
import com.gvsoft.analyse.inf.IOrder;
import com.gvsoft.analyse.inf.IPacket;
import com.gvsoft.analyse.order.ErrorOrder;
import com.gvsoft.analyse.order.TokenOrder;
import com.gvsoft.analyse.packet.LoginPacket;
import com.gvsoft.common.CommonTools;
import com.gvsoft.conns.ConnTimeoutCleanThread;

import java.io.*;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.UUID;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import com.gvsoft.conns.GVConnTools;
import com.gvsoft.conns.GVConnection;
import com.gvsoft.msgqueue.GVQueue;
import com.gvsoft.msgqueue.CQueueConsumer;
import com.gvsoft.msgqueue.MQueueConsumer;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

/**
 * Created with IntelliJ IDEA.
 * ProjectName:gvMsgRouting
 * Author: zhaoqiubo
 * Date: 15/7/21
 * Time: 上午9:24
 * Desc: 通讯服务主线程
 */
public class GVServer implements Runnable {

    //事件选择器
    private static Selector selector;
    //定义包头的字节大小
    private static final int PACKET_HEAD_LENGTH = 4;
    //目前我理解ServerSocketChannel就是SocketChannel通道的容器
    private ServerSocketChannel serverSocketChannel;
    //断包处理，前一截包含完整包头；
    private String endPacketStr = "";
    //断包处理，前一截不包含完整包头；
    private String endBufferStr = "";
    //断包处理，前一截包含完整包头时，包体的大小标记；
    private int remainBodySize = 0;

    private static Logger logger = LogManager.getLogger(GVServer.class.getName());

    public static void main(String arg[]) throws Exception {
        //初始化配置数据
        Config cfg = new Config(arg[0]);
        final GVServer gvServer = new GVServer();
        //启动ServerSocket通道
        gvServer.initServer(cfg);

        ExecutorService threadPool = Executors.newCachedThreadPool();
        //启动通讯服务主线程
        threadPool.execute(gvServer);
        //启动通讯报文消费线程
        threadPool.execute(new CQueueConsumer(cfg.getWaitTime()));
        //启动超时通道、失效通道回收线程
        threadPool.execute(new ConnTimeoutCleanThread(cfg.getCleanThreadOutTime(), cfg.getCleanThreadCycle()));
        threadPool.execute(new MQueueConsumer());
    }

    /**
     * 初始化NIOServer各种参数，并打开服务等待客户端连接
     *
     * @param cfg
     * @throws IOException
     */
    private void initServer(Config cfg) throws IOException {
        try {
            //打开一个serversocket通道，ServerSocketChannel是一个监控是否有新连接进入的通道。
            serverSocketChannel = ServerSocketChannel.open();
            //将这个serversokect通道设置为非阻塞模式
            serverSocketChannel.configureBlocking(false);
            //绑定serversokect的ip和端口
            serverSocketChannel.socket().bind(new InetSocketAddress(cfg.getIp(), cfg.getPort()));
            //打开选择器
            selector = Selector.open();
            //将此通道注册给选择器selector
            serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
            logger.info("GVServer主通讯服务已经启动IP：" + cfg.getIp() + "，端口：" + cfg.getPort() + "，开始接收客户端连接…………");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public void run() {

        while (true) {
            try {

                //监听事件key
                selector.select(2000);
                //迭代一组事件key
                Iterator<SelectionKey> keys = selector.selectedKeys().iterator();
                while (keys.hasNext()) {
                    //定义一个socket通道
                    SocketChannel socketChannel = null;

                    int count = 0;

                    SelectionKey key = keys.next();
                    //  Logs.info("有网络事件被触发，事件类型为：" + key.interestOps());
                    //删除Iterator中的当前key，避免重复处理
                    keys.remove();
                    if (!key.isValid()) {
                        continue;
                    } else if (key.isAcceptable()) {
                        //从客户端送来的key中获取ServerSocket通道
                        serverSocketChannel = (ServerSocketChannel) key.channel();
                        //接收此ServerSocket通道中的Socket通道，accept是一个阻塞方法，一直到获取到连接才会继续
                        socketChannel = serverSocketChannel.accept();
                        //将此socket通道设置为非阻塞模式
                        socketChannel.configureBlocking(false);
                        //将此通道注册到selector，并等待接收客户端的读入数据
                        socketChannel.register(selector, SelectionKey.OP_READ);
                        allocToken(socketChannel);

                    } else if (key.isReadable()) {

                        //获取事件key中的channel
                        socketChannel = (SocketChannel) key.channel();
                        ByteBuffer byteBuffer = ByteBuffer.allocate(Config.getReadBlock());
                        //清理缓冲区，便于使用
                        byteBuffer.clear();
                        //将channel中的字节流读入缓冲区
                        count = socketChannel.read(byteBuffer);
                        byteBuffer.flip();
                        //处理粘包
                        if (count > 0) {
                            try {
                                handlePacket(socketChannel, byteBuffer);
                            } catch (Exception e) {
                                e.printStackTrace();
//                                continue;//如果当前包存在非法抛出异常，那么不再进行处理直接跳出循环，处理下一个包；此处存疑，测试阶段暂时注释
                            }
                        } else if (count == 0) {
                            continue;
                        } else {
                            socketChannel.close();

                        }

                    } else if (key.isWritable()) {
                        ((SocketChannel) key.channel()).register(selector, SelectionKey.OP_READ);
                    }
                }

            } catch (IOException e) {
                e.printStackTrace();
            }

        }
    }

    /**
     * 处理断包和粘包现象
     * @param socketChannel
     * @param byteBuffer
     */
    private void handlePacket(SocketChannel socketChannel, ByteBuffer byteBuffer) {
        //标记读取缓冲区起始位置
        int location = 0;
        //如果缓冲区从0到limit的数量大于包体大小标记数字
        while (byteBuffer.remaining() > PACKET_HEAD_LENGTH) {
            //包体大小标记
            String strBsize;
            //如果endPacket的字节length大于0，则证明：断包的前一截为包含包头和包体的；
            if (endPacketStr.getBytes().length > 0) {

                String strPacket = endPacketStr.substring(PACKET_HEAD_LENGTH) + new String(byteBuffer.array(), 0, remainBodySize);
                byteBuffer.position(remainBodySize);
                location = remainBodySize;
//                                    if(logger.isDebugEnabled()) {
                logger.info("【断包处理】(包含包体)合并后的报文:" + strPacket + "，缓冲区的position:" + location);
//                                    }
                offerPacket(socketChannel, strPacket);
                //处理完毕，清理断包的前一截，以便于下次使用；
                endPacketStr = "";
                //清理后一截报文的字节数标记；
                remainBodySize = 0;
                continue;
            //如果endBufferStr的字节length大于0，则证明：断包的前一截仅包含包头或包头的一部分，不包含包体；
            } else if (endBufferStr.getBytes().length>0){

                strBsize = (new StringBuffer(endBufferStr).append(new
                        String(byteBuffer.array(), location, PACKET_HEAD_LENGTH - endBufferStr.getBytes().length))).toString();

                //移动缓冲区position
                byteBuffer.position(PACKET_HEAD_LENGTH - endBufferStr.getBytes().length);
                location = byteBuffer.position();
                //得到包体大小
                int byteBufferSize = Integer.parseInt(strBsize.trim());
                //进行报文合并，把保存的仅包含包头或包头一部分的前一截与后一截合并
                String strPacket = endBufferStr+(new String(byteBuffer.array(), PACKET_HEAD_LENGTH - endBufferStr.getBytes().length, byteBufferSize));
                           byteBuffer.position(location  + byteBufferSize);//将缓冲区的位置移动到下一个包体大小标记位置
                location = byteBuffer.position();
                logger.info("【断包处理】(不包含包体)合并后的报文:" + strPacket + "，缓冲区的position:" + location);
                offerPacket(socketChannel, strPacket);
                endBufferStr="";
                continue;
            //进入正常处理（规范的报文处理，不考虑断包）
            }else{
                strBsize = new String(byteBuffer.array(), location, PACKET_HEAD_LENGTH);
                //移动缓冲区position
                byteBuffer.position(location + PACKET_HEAD_LENGTH);
            }
            if (logger.isDebugEnabled()) {
                logger.debug("收到客户端包体大小：" + strBsize + "，查看position变化：" + byteBuffer.position());
            }
            //得到包体大小
            int byteBufferSize = Integer.parseInt(strBsize.trim());
            //如果从缓冲区当前位置到limit大于包体大小，证明粘包了，进行包体处理。等于则为正常包体，不存在粘包现象。
            if (byteBuffer.remaining() >= byteBufferSize) {

                String strPacket = endBufferStr+(new String(byteBuffer.array(), PACKET_HEAD_LENGTH + location, byteBufferSize));
                byteBuffer.position(location + PACKET_HEAD_LENGTH + byteBufferSize);//将缓冲区的位置移动到下一个包体大小标记位置
                if (logger.isDebugEnabled()) {
                    logger.debug("收到客户端包体内容：" + strPacket + "，2查看position变化：" + byteBuffer.position());
                }
                //将字符串报文封装为类
                offerPacket(socketChannel, strPacket);
                location = byteBuffer.position();//设定读取缓冲区起始位置
            //如果缓冲区当前位置到limit小于包体，证明断包了,进行断包处理
            } else {

                endPacketStr = new String(byteBuffer.array(), location, byteBuffer.limit() - location);
                remainBodySize = Integer.parseInt(endPacketStr.substring(0,4).trim())-endPacketStr.getBytes().length+PACKET_HEAD_LENGTH;
                //已经找到断包前半截，所以把整个buffer的position调整至最后，不再处理。等待新的key进入
                byteBuffer.position(byteBuffer.limit());
                logger.info("处理断包仅包含完整包头的尾部报文，缓冲区位置:" + location + "，缓冲区limit:" + byteBuffer.limit() + "，包含完全包头的剩余字符:" + endPacketStr + "，bodySize:" + remainBodySize);

            }
        }
        //处理仅包含包头前一截的报文；
        if (byteBuffer.remaining() > 0) {

            //缓冲区中剩余的仅包含包头前一截的报文
            endBufferStr = new String(byteBuffer.array(), location, byteBuffer.limit() - location);

            logger.info("处理断包仅包含包头前一截的尾部报文，缓冲区位置:" + location + "，缓冲区limit:" + byteBuffer.limit() + "，不包含完全包头的剩余字符：" + endBufferStr);
            //移动缓冲区指针到最后，代表已经保存了前一截报文，无需再进行处理；
            byteBuffer.position(byteBuffer.limit());
        }
        //我也不知道这是否有用，能不能释放内存资源
        byteBuffer.clear();
    }

    /**
     * 处理不合法报文以及将合法报文放入队列等待处理
     * @param socketChannel
     * @param strPacket
     */
    private void offerPacket(SocketChannel socketChannel, String strPacket) {
        IPacket packet = AnalyseTools.analysePacket(strPacket);
        if (packet.getHeader().equals(LoginPacket.HEADER)) {
            handleLoginPacket(socketChannel, packet);
        }
        //如果类为空或者从handle单例map中无法取到类，则证明报文非法
        if (packet == null || Config.getPacketInstance(packet.getHeader()) == null) {
            //不在服务端识别范围内的报文，回复E响应，告知客户端不合法
            ErrorOrder errorOrder = (ErrorOrder) Config.getOrderInstance(ErrorOrder.HEADER);
            errorOrder.initErrorOrder(errorOrder.INVAILD_REQ_CODE, errorOrder.INVAILD_REQ_MSG);

            logger.info("客户端发送非法报文：" + strPacket);
            GVServer.write2Client(errorOrder, socketChannel);
            //将合法报文放入消息队列中等待处理
        } else {
            if (!GVQueue.CQUEUE.offer(packet)) {
                logger.error("消息队列已满，请增加队列容量！");

            } else {
                if (logger.isDebugEnabled()) {
                    logger.debug("将添加至消息队列，队列内消息数量为：" + GVQueue.CQUEUE.size());
                }
            }
        }
    }

   /* *//**
     * 处理从通道中读入的包，包括：1、粘包的处理（断包目前还未处理）；2、非法包处理；
     *
     * @param socketChannel
     * @param byteBuffer
     *//*
    private void handleRead(SocketChannel socketChannel, ByteBuffer byteBuffer) {
        //标记读取缓冲区起始位置
        int location = 0;
        //如果缓冲区从0到limit的数量大于包体大小标记数字
        while (byteBuffer.remaining() > PACKET_HEAD_LENGTH) {
            //包体大小标记
            String strBsize;
            strBsize = new String(byteBuffer.array(), location, PACKET_HEAD_LENGTH);
            //移动缓冲区position
            byteBuffer.position(location + PACKET_HEAD_LENGTH);
            if (logger.isDebugEnabled()) {
                logger.debug("收到客户端包体大小：" + strBsize + "，查看position变化：" + byteBuffer.position());
            }
            //得到包体大小
            int byteBufferSize = Integer.parseInt(strBsize.trim());
            //如果从缓冲区当前位置到limit大于包体大小，证明粘包了，进行包体处理
            if (byteBuffer.remaining() >= byteBufferSize) {

                String strPacket = new String(byteBuffer.array(), PACKET_HEAD_LENGTH + location, byteBufferSize);
                byteBuffer.position(location + PACKET_HEAD_LENGTH + byteBufferSize);//将缓冲区的位置移动到下一个包体大小标记位置
                if (logger.isDebugEnabled()) {
                    logger.debug("收到客户端包体内容：" + strPacket + "，2查看position变化：" + byteBuffer.position());
                }
                handlePacket2(socketChannel, strPacket);

                location = location + PACKET_HEAD_LENGTH + byteBufferSize;//设定读取缓冲区起始位置
            }
        }
        //我也不知道这是否有用，能不能释放内存资源
        byteBuffer.clear();
    }*/


    /**
     * 专门处理客户端登录报文，保存GVConn到缓冲区
     * 【注】对于userId重复的情况，在这里不做处理了，由业务系统自己处理，
     * 这里对userId重复相当于后登录的用户替换了先登录用户的通道。
     *
     * @param socketChannel
     * @param packet
     */
    private void handleLoginPacket(SocketChannel socketChannel, IPacket packet) {

        GVConnection gvConn = new GVConnection(packet.getClientToken(),
                packet.getPacketBody(), socketChannel, CommonTools.systemTimeUtc());
        GVConnTools.addConn2Cache(gvConn);

    }

    /**
     * 给连接上来的通道分配token，并将通道存入缓冲区
     *
     * @param socketChannel
     */
    private void allocToken(SocketChannel socketChannel) {

        //给连接上来的通道分配token
        TokenOrder tokenOrder = (TokenOrder) Config.getOrderInstance(TokenOrder.HEADER);
        String token = UUID.randomUUID().toString();
        tokenOrder.initTokenOrder(token);
        //返回给客户端token
        write2Client(tokenOrder, socketChannel);
        logger.info("客户端:<" + token + ">已经连接！");
        //将连接后的channel保存到全局map中
        //2015.8.13修改，先把userId存为null，等待用户登录后，在将，
//        GVConnection gvConnection = new GVConnection(token,null, socketChannel, CommonTools.systemTimeUtc());
//        GVConnTools.addConn2Cache(gvConnection);

    }

    /**
     * 向客户端写入信息的方法
     *
     * @param iOrder        报文处理类接口
     * @param socketChannel
     */
    public static synchronized void write2Client(IOrder iOrder, SocketChannel socketChannel) {
        try {
            socketChannel.register(selector, SelectionKey.OP_WRITE);
            //创建一个byteBuffer用来存储要写入的buffer
            ByteBuffer byteBuffer = ByteBuffer.allocate(Config.getWriteBlock());
            //得出整个包体的长度
            String packetSize = Integer.toString(iOrder.generateOrderStr().getBytes().length);
            //讲包体长度放入buffer的前四位
            byteBuffer.put(packetSize.getBytes());
            //移动buffer的postion指针到第四位，包体将从第四位开始写入
            byteBuffer.position(PACKET_HEAD_LENGTH);
            String str = iOrder.generateOrderStr();
            //写入包体
            if (logger.isDebugEnabled()) {
                logger.debug("服务端写入通道的包体：" + str);
            }
            byteBuffer.put(str.getBytes());
            byteBuffer.flip();
            socketChannel.write(byteBuffer);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

}
